#include "libcomm.h"

ThreadPool::WorkerThread::WorkerThread(ThreadPool* pool, int index, const char* name)
: m_pool(pool), m_index(index), Thread(name) {}

int ThreadPool::WorkerThread::Process()
{
	return m_pool->Process(m_index);
}

class DefaultThreadPool : public ThreadPool
{
public:
	ThreadPool::ThreadPoolProc m_proc;
	CommType m_arg;
	DefaultThreadPool(ThreadPool::ThreadPoolProc proc, const CommType& arg, int threadCount, const char* poolName)
		:ThreadPool(threadCount,poolName),m_proc(proc),m_arg(arg){}
	virtual int Process(int index)
	{
		return m_proc(index,m_arg);
	}
};

ThreadPool* ThreadPool::Start(ThreadPoolProc proc, const CommType& arg, int threadCount, const char* poolName)
{
	MYLIB_ASSERT_NOTNULL(proc);
	DefaultThreadPool* pool = new DefaultThreadPool(proc,arg,threadCount,poolName);
	MYLIB_ASSERT_NOTNULL(pool);
	pool->Start();
	return static_cast<ThreadPool*>(pool);
}

//////////////////////////////////////////////////////////////////////////

ThreadPool::ThreadPool(int threadCount, const char* poolName)
	:m_name(poolName),m_started(false),m_threadCount(threadCount)
{
	if(m_name.empty())
	{
		static int index = 0;
		m_name = String::format("trdpool%03d",++index);
	}
	
	m_threadList = (Thread**)Buffer::calloc(threadCount,sizeof(Thread*));
	MYLIB_ASSERT_NOTNULL(m_threadList);
	for (int i=0;i<threadCount;i++)
	{
		Thread* trd = new WorkerThread(this,i,String(64).format("%s_trd%03d",(char*)m_name,i+1));
		MYLIB_ASSERT_NOTNULL(trd);
		m_threadList[i] = trd;
	}
}

ThreadPool::~ThreadPool(void)
{
	ThreadLockGuard guard(&m_threadLock);
	Stop(true);
	for (int i=0;i<m_threadCount;i++)
	{
		Thread* trd = m_threadList[i];
		delete trd;
	}
	Buffer::free(m_threadList);
}

Thread* ThreadPool::operator[](int index)
{
	ThreadLockGuard guard(&m_threadLock);
	MYLIB_ASSERT(index>=0 && index<m_threadCount);
	return m_threadList[index];
}

const char* ThreadPool::Name() const
{
	return m_name;
}

void ThreadPool::Start()
{
	ThreadLockGuard guard(&m_threadLock);
	if(m_started)
		return;
	for (int i=0;i<m_threadCount;i++)
	{
		Thread* trd = m_threadList[i];
		MYLIB_ASSERT_NOTNULL(trd);
		trd->Run();
	}
	m_started = true;
}

void ThreadPool::Stop(bool cancelThread)
{
	ThreadLockGuard guard(&m_threadLock);
	if(!m_started)
		return;
	if(cancelThread)
	{
		for (int i=0;i<m_threadCount;i++)
		{
			Thread* trd = m_threadList[i];
			MYLIB_ASSERT_NOTNULL(trd);
			trd->Terminate();
		}
		Thread::Sleep(200);// waiting for threads to clear up, eg:__wait_cleanup.
	}
	
	for (int i=0;i<m_threadCount;i++)
	{
		Thread* trd = m_threadList[i];
		MYLIB_ASSERT_NOTNULL(trd);
		trd->Join();
	}
	m_started = false;
}
